package skewreduce.framework;

import java.io.File;
import java.io.IOException;
import java.lang.reflect.*;
import java.net.URL;
import java.util.Properties;

import org.apache.log4j.PropertyConfigurator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DataInputBuffer;
import org.apache.hadoop.io.DataOutputBuffer;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Progress;
import org.apache.hadoop.util.ReflectionUtils;

/**
 * fake runner for local debugging
 */
public class FakeMergeRunner {
    static class BogusStatusReporter extends StatusReporter {
		Counters counters = new Counters();
		
		@Override
		public Counter getCounter(Enum<?> key) {
			return counters.findCounter(key);
		}

		@Override
		public Counter getCounter(String arg0, String arg1) {
			return counters.findCounter(arg0,arg1);
		}

		@Override
		public void progress() {
			// do nothing
		}

		@Override
		public void setStatus(String arg0) {
			// do nothing
		}
	}

    static final String LOG4J_PROP = "log4j.properties";
    static final String LOG_CONFIG_ATTR = "skewreduce.log4j";

    private static void configureLog(Configuration conf) {
        String log4j = conf.get(LOG_CONFIG_ATTR);
        File f = null;

        if ( log4j == null ) {
            // search logging file in current directory
            f = new File(LOG4J_PROP);
        } else {
            f = new File(log4j);
        }
        if ( f.exists() ) {
            System.err.println("Setting up logging from "+f.getPath());
            PropertyConfigurator.configure(f.getPath());
            return;
        } else {
            // try to load from local resource bundle
            URL url = EaggDriver.class.getResource("/skewreduce/"+LOG4J_PROP);
            if ( url != null ) {
                System.err.println("Setting up logging from default properties");
                PropertyConfigurator.configure(url);
                return;
            }
        }

        System.err.println("Setting up basic logging");

        // configure a console
        Properties props = new Properties();
        props.setProperty("log4j.rootLogger","INFO, CONSOLE");
        props.setProperty("log4j.appender.CONSOLE","org.apache.log4j.ConsoleAppender");
        props.setProperty("log4j.appender.CONSOLE.layout","org.apache.log4j.PatternLayout");
        props.setProperty("log4j.appender.CONSOLE.layout.ConversionPattern","%d [%t] %-5p %c - %m%n");
        props.setProperty("log4j.appender.CONSOLE.target","System.err");

        PropertyConfigurator.configure(props);
    }

 
	@SuppressWarnings("unchecked")
	public static <INKEY,INVALUE,OUTKEY,OUTVALUE> void main(String[] args) throws Exception {
        GenericOptionsParser options = new GenericOptionsParser(new Configuration(),args);
        Configuration conf = options.getConfiguration();
        
        String[] rest = options.getRemainingArgs();
        String inputStr = null;
        String outputStr = "./output-tmp";
        boolean newMerge = false;
        
        for ( int i = 0; i < rest.length; ++i ) {
        	if ( rest[i].equals("-input") ) {
        		inputStr = rest[++i];
        	} else if ( rest[i].equals("-output") ) {
        		outputStr = rest[++i];
        	} else if ( rest[i].equals("-newMerge") ) {
                newMerge = true;
            }
        }

        conf.set("fs.default.name","file:///");
        conf.set("mapred.output.dir", outputStr);
        
        configureLog(conf);

        Job job = ( newMerge ) ? skewreduce.dfof.dFoFMerge2.getJobInstance(conf) : skewreduce.dfof.dFoFMerge.getJobInstance(conf);
        Configuration jobConf = job.getConfiguration();

        TaskAttemptID attemptID = new TaskAttemptID("local",0,true,0,0);
        TaskAttemptContext attemptContext = new TaskAttemptContext(jobConf,attemptID);
        // prepare context
        Reducer<INKEY,INVALUE,OUTKEY,OUTVALUE> reducer = (Reducer<INKEY,INVALUE,OUTKEY,OUTVALUE>)ReflectionUtils.newInstance(attemptContext.getReducerClass(),jobConf);

        // assuming all required classes are already in the classpath
        Path inputPath = new Path(inputStr);
        Path tmpPath = new Path("./tmp-yongchul");

        FileSystem localFs = FileSystem.getLocal(jobConf);
        localFs.delete(new Path(outputStr),true);
        localFs.delete(tmpPath,true);

        // output -- set to local output
        // status reporter

        SequenceFile.Sorter sorter = new SequenceFile.Sorter(localFs,job.getGroupingComparator(),job.getMapOutputKeyClass(),job.getMapOutputValueClass(),jobConf);

        final SequenceFile.Sorter.RawKeyValueIterator iter = sorter.merge(new Path[] { inputPath },false,2,tmpPath);
        org.apache.hadoop.mapred.RawKeyValueIterator kviter = new org.apache.hadoop.mapred.RawKeyValueIterator() {
            DataInputBuffer keyBuf = new DataInputBuffer();
            DataInputBuffer valBuf = new DataInputBuffer();
            DataOutputBuffer rawValBuf = new DataOutputBuffer(1024);

            public void close() throws IOException {
                iter.close();
            }
            public DataInputBuffer getKey() throws IOException {
                return keyBuf;
            }
            public DataInputBuffer getValue() throws IOException {
                return valBuf;
            }
            public Progress getProgress() {
                return iter.getProgress();
            }
            public boolean next() throws IOException {
                boolean hasNext = iter.next();
                if ( hasNext ) {
                    DataOutputBuffer obuf = iter.getKey();
                    keyBuf.reset(obuf.getData(),0,obuf.getLength());
                    rawValBuf.reset();
                    iter.getValue().writeUncompressedBytes(rawValBuf);
                    valBuf.reset(rawValBuf.getData(),0,rawValBuf.getLength());
                }
                return hasNext;
            }
        };


        OutputFormat outputFormat = ReflectionUtils.newInstance(attemptContext.getOutputFormatClass(),jobConf);
        OutputCommitter committer = outputFormat.getOutputCommitter(attemptContext);

        RecordWriter<OUTKEY,OUTVALUE> output = outputFormat.getRecordWriter(attemptContext);

        Constructor<Reducer.Context> contextConstructor =
            Reducer.Context.class.getConstructor
            (new Class[]{org.apache.hadoop.mapreduce.Reducer.class,
                         Configuration.class,
                         org.apache.hadoop.mapreduce.TaskAttemptID.class,
                         org.apache.hadoop.mapred.RawKeyValueIterator.class,
                         org.apache.hadoop.mapreduce.Counter.class,
                         org.apache.hadoop.mapreduce.RecordWriter.class,
                         org.apache.hadoop.mapreduce.OutputCommitter.class,
                         org.apache.hadoop.mapreduce.StatusReporter.class,
                         org.apache.hadoop.io.RawComparator.class,
                         Class.class,
                         Class.class});

        Reducer<INKEY,INVALUE,OUTKEY,OUTVALUE>.Context reducerContext = null;

        BogusStatusReporter reporter = new BogusStatusReporter();
        
        reducerContext = contextConstructor.newInstance(reducer,
                jobConf,
                attemptID,
                kviter,
                reporter.getCounter("org.apache.hadoop.mapred.Task.Counter","REDUCE_INPUT_RECORDS"),
                output,
                committer,
                reporter,
                job.getGroupingComparator(),
                job.getMapOutputKeyClass(),
                job.getMapOutputValueClass()
        );

        reducer.run(reducerContext);
        try {
            iter.close();
        } catch ( IOException ignore ) {}

        if ( committer.needsTaskCommit(attemptContext) ) {
            committer.commitTask(attemptContext);
        }
        output.close(reducerContext);
	}
}
